package net.bwie.realtime.vehicle.dwd.function;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

import org.apache.flink.types.Row;

public class MysqlCustomSink extends RichSinkFunction<Row> {
    // 数据库连接、预编译语句
    private Connection conn;
    private PreparedStatement pstmt;

    // 初始化连接（在任务启动时执行一次）
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 加载 MySQL 驱动（8.0.x 版本使用 com.mysql.cj.jdbc.Driver）
        Class.forName("com.mysql.cj.jdbc.Driver");
        // 连接数据库（替换为你的 MySQL 地址、库名、用户名、密码）
        conn = DriverManager.getConnection(
                "jdbc:mysql://localhost:3306/vehicle_db?useSSL=false&serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=utf8",
                "root",
                "123456"
        );
        // 预编译插入语句（根据你的 MySQL 表结构调整字段和占位符）
        // 表结构：monitorId_avgspeed_carcount (start_time BIGINT, end_time BIGINT, monitor_id STRING, avg_speed DOUBLE, car_count INT, rn INT, PRIMARY KEY(start_time, monitor_id))
        String sql = "INSERT INTO monitorId_avgspeed_carcount " +
                "(start_time, end_time, monitor_id, avg_speed, car_count, rn) " +
                "VALUES (?, ?, ?, ?, ?, ?) " +
                "ON DUPLICATE KEY UPDATE " +  // 处理主键冲突（更新已有数据）
                "end_time = VALUES(end_time), avg_speed = VALUES(avg_speed), " +
                "car_count = VALUES(car_count), rn = VALUES(rn)";
        pstmt = conn.prepareStatement(sql);
    }

    // 每条数据触发一次（写入逻辑）
    @Override
    public void invoke(Row value, Context context) throws Exception {
        // Row 中的字段顺序与 SQL 查询结果一致：window_start, window_end, monitorId, avg_speed, car_count, rn
        pstmt.setLong(1, (Long) value.getField(0));   // start_time（window_start）
        pstmt.setLong(2, (Long) value.getField(1));   // end_time（window_end）
        pstmt.setString(3, (String) value.getField(2)); // monitor_id
        pstmt.setDouble(4, (Double) value.getField(3)); // avg_speed
        pstmt.setInt(5, (Integer) value.getField(4));   // car_count
        pstmt.setInt(6, (Integer) value.getField(5));   // rn
        // 执行写入
        pstmt.executeUpdate();
    }

    // 任务结束时关闭资源
    @Override
    public void close() throws Exception {
        super.close();
        if (pstmt != null) {
            pstmt.close();
        };
        if (conn != null) {
            conn.close();
        };
    }
}